/**
 * Copyright (C), 2018 store
 * Encoding: UTF-8
 * Date: 19-9-16 上午11:38
 * History:
 */
package com.swak.frame.ratelimit.resolver;

import java.math.BigDecimal;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;

import com.google.common.util.concurrent.RateLimiter;
import com.swak.frame.common.SpringBeanFactory;
import com.swak.frame.config.AbstractStateConfigurable;
import com.swak.frame.ratelimit.LimitType;
import com.swak.frame.ratelimit.SwakRateLimiter;
import com.swak.frame.ratelimit.config.LimitCommand;
import com.swak.frame.ratelimit.event.RateAlarmEvent;
import com.swak.frame.ratelimit.event.RateLimitEvent;
import com.swak.frame.ratelimit.event.RateLimitEventInfo;
import com.swak.frame.util.ObjectUtil;

/**
 * 令牌桶算法，对资源进行限流 ClassName: LocalTokenLimitRater.java
 * 
 * @author colley.ma
 * @date 2021年3月26日 下午3:17:53
 */
public class LocalTokenLimitRater extends AbstractStateConfigurable<LimitCommand> implements ApplicationListener<RateLimitEvent>,SwakRateLimiter {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private Map<String, RateLimiter> limiterConfig = new ConcurrentHashMap<String, RateLimiter>();

	public LocalTokenLimitRater() {
		super(LimitCommand.class);
	}

	@Override
	public boolean tryAcquire(String resource) {
		if (StringUtils.isEmpty(resource)) {
			return true;
		}
		RateLimiter limiter = limiterConfig.get(resource);
		if (!ObjectUtil.nonNull(limiter)) {
			return true;
		}
		boolean acquire = limiter.tryAcquire();
		if (!acquire) {
			LimitCommand config = getConfig().get(resource);
			if (config != null && config.getEvent()!=null) {
				RateAlarmEvent alarmEvent = SpringBeanFactory.getBean(config.getEvent());
				if (alarmEvent != null) {
					RateLimitEventInfo eventInfo = RateLimitEventInfo.newRateLimitEventInfo(resource, config,LimitType.LOCAL_TOKEN);
					alarmEvent.event(eventInfo);
				}
			}
		}
		return acquire;
	}

	/**
	 * 初始化令牌桶
	 * 
	 * @param resource
	 * @param qps
	 */
	private RateLimiter updateResourceQps(String resource) {
		LimitCommand config = getConfig().getOrDefault(resource, new LimitCommand(false));
		// 小于等于0,不限流
		if (!config.isEnable() || config.getQps() <= 0) {
			if (limiterConfig.containsKey(resource)) {
				limiterConfig.remove(resource);
			}
			logger.info("接口：{},限流已关闭", resource);
			return null;
		}
		
		RateLimiter rateLimiter = limiterConfig.get(resource);
		if ((rateLimiter != null)) {
			rateLimiter.setRate(config.getQps());
			return rateLimiter;
		}
		rateLimiter = RateLimiter.create(config.getQps());
		limiterConfig.put(resource, rateLimiter);
		return rateLimiter;
	}

	@Override
	public void onApplicationEvent(RateLimitEvent event) {
		LimitCommand limitConfig = event.getConfig();
		// 本地限流触发
		if (LimitType.LOCAL_TOKEN.equals(event.getLimitType())) {
			if (limitConfig != null) {
				computeResourceQps(event.getResource(), limitConfig);
				getConfig().put(event.getResource(), limitConfig);
				updateResourceQps(event.getResource());
			}
		}
	}
	
	/**
	 * 动态计算QPS
	 * @param resource
	 * @param limitConfig
	 */
	private void computeResourceQps(String resource,LimitCommand  limitConfig) {
		//传递配置的qps和实例数计算每个实例的qps
		if(!limitConfig.isDynamic() || limitConfig.getInstances()<=1) {
			limitConfig.setDynamic(false);  //未开启或则实例为1，不处理
			return;
		}
		//qps配置小于机器数,默认配置数为qps
		if(limitConfig.isDynamic() && limitConfig.getQps()<=limitConfig.getInstances()) {
			limitConfig.setDynamic(false);  //未开启或则实例为1，不处理
			return;
		}
		BigDecimal qpsDecimal = new BigDecimal(limitConfig.getQps()).divide(new BigDecimal(limitConfig.getInstances()), 4);
		int qps = Math.max(qpsDecimal.intValue(),10); 
		limitConfig.setQps(qps);
	}
}
